package com.huatai.datacenter.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.huatai.common.api.R;
import com.huatai.common.util.ObjectUtil;
import com.huatai.datacenter.entity.DataServicesEntity;
import com.huatai.datacenter.entity.RabbitMQEntity;
import com.huatai.datacenter.entity.TopicTaskEntity;
import com.huatai.datacenter.entity.datamodel.DataModelFieldEntity;
import com.huatai.datacenter.job.KafkaConsumerListener;
import com.huatai.datacenter.mapper.RabbitMQMapper;
import com.huatai.datacenter.utils.RedisUtil;
import com.huatai.dataservice.dubbo.DataDisposeDubboAPI;
import com.rabbitmq.client.*;
import lombok.extern.log4j.Log4j;
import org.apache.commons.lang.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.conn.HttpHostConnectException;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;


/**
 * @author Lion
 * @date 2023/8/1  10:47
 */
@Service
@Log4j
public class RabbitMQService extends BaseService<RabbitMQMapper, RabbitMQEntity> {

	@Autowired
	private DataModelFieldService dataModelFieldService;

	@DubboReference
	private DataDisposeDubboAPI dataDisposeDubbo;

	@Autowired
	private TopicTaskService topicTaskService;

	@Autowired
	private KafkaConsumerListener kafkaConsumerListener;

	@Autowired
	private DataServices dataServices;

	public static Map<String,Object> map=new ConcurrentHashMap<>();




	public int getConnection(RabbitMQEntity rabbitMQEntity){

		ConnectionFactory factory = new ConnectionFactory();
		//服务地址
		factory.setHost(rabbitMQEntity.getHost());
		//服务所在端口，不填默认5672
		factory.setPort(rabbitMQEntity.getPort());
		//管理员账号
		factory.setUsername(rabbitMQEntity.getUsername());
		//管理员密码
		factory.setPassword(rabbitMQEntity.getPassword());
		//虚拟机
		factory.setVirtualHost(rabbitMQEntity.getVirtualHost());
		try {
			factory.newConnection();
			return 1;
		} catch (IOException e) {
			return 0;
		} catch (TimeoutException e) {
			return 0;
		}
	}

	/**
	 * 添加rabbitMq任务并启动监听
	 * @param entity
	 * @return
	 */
	public R createRabbitMqTask(TopicTaskEntity entity){
		List<DataModelFieldEntity> dataModelFieldEntities = dataModelFieldService.getDataModelFieldList(entity.getModelId());
		dataDisposeDubbo.existTable(entity.getDataTable(),jsonStructureData(dataModelFieldEntities));
		//获取到mq集群信息
		RabbitMQEntity mqEntity = this.getById(entity.getClusterId());
		//创建获取连接
		Connection connection = getConnection2(mqEntity);
		if (ObjectUtil.isNotEmpty(connection)) {
			topicTaskService.save(entity);
			//消费消息
			Thread thread = new Thread(()->asyncGet(mqEntity, connection, entity));
			//判断模型id和质量是否存在，如果存在，则写入一条记录到记录表中
			thread.start();
			topicTaskService.saveUseModelData(entity);
			return R.data("添加成功！");
		}
		return R.data("连接失败！");
	}

	public void asyncGet(RabbitMQEntity mqEntity, Connection connection,TopicTaskEntity entity) {
			try {
				//当前工作线程名称
				String workerName = Thread.currentThread().getName();
				Channel channel = connection.createChannel();
				String exchangeName = mqEntity.getExchangeName();
				// 1.交换机 2.发布模式  3持久化
				channel.exchangeDeclare(exchangeName, "direct", true);
				String routingName = entity.getTopicName();
				//创建一个与路由同名的队列  1队列名称 2 是否持久化 3是否排他 4 是否自动删除 5其他属性
				channel.queueDeclare(routingName,true,false,false,null);
				channel.queueBind(routingName, exchangeName,routingName);
				channel.basicConsume(routingName, true, new DefaultConsumer(channel) {
					//启动成功
					@Override
					public void handleConsumeOk(String consumerTag) {
						System.out.printf("消费者%s - 启动成功%n", workerName);
						long id = Thread.currentThread().getId();
						map.put(entity.getId().toString(),id);
						System.out.println("内层id"+id);
					}

					//消费消息
					@Override
					public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
						System.out.printf("消息者%s - 获取消息：%s,路由：%s%n", workerName, new String(body), envelope.getRoutingKey());
						//保存实时数据到mysql数据库中
						saveRealTimeData(new String(body),entity.getModelId(),entity.getModelName(),entity.getDataTable()
							,entity.getStructure(),entity.getQualityRule(),entity.getTopicName());
					}
				});
			} catch (IOException e) {
				throw new RuntimeException(e);
			}
	}


	/**
	 * 获取mq连接
	 * @param mqEntity
	 * @return
	 */
	public Connection getConnection2(RabbitMQEntity mqEntity) {
		ConnectionFactory factory = new ConnectionFactory();
		//服务地址
		factory.setHost(mqEntity.getHost());
		//服务所在端口，不填默认5672
		factory.setPort(mqEntity.getPort());
		//管理员账号
		factory.setUsername(mqEntity.getUsername());
		//管理员密码
		factory.setPassword(mqEntity.getPassword());
		//虚拟机
		factory.setVirtualHost(mqEntity.getVirtualHost());
		//设置连接超时时间5秒
		factory.setConnectionTimeout(50000);
		try {
			Connection connection = factory.newConnection();
			if (connection!=null){
				mqEntity.setStatus(1);
				this.updateById(mqEntity);
				return connection;
			}else {
				mqEntity.setStatus(0);
				this.updateById(mqEntity);
			}
		} catch (Exception e) {
			mqEntity.setStatus(0);
			this.updateById(mqEntity);
			//throw new RuntimeException(e);
			log.error("rabbitmq连接异常{}"+e.getMessage());
		}
		return null;
	}

	public List<JSONObject> jsonStructureData(List<DataModelFieldEntity> dataModelFieldEntities){

		List<JSONObject> jsonObjects = new ArrayList<>();
		for (int i=0;i< dataModelFieldEntities.size();i++){
			JSONObject jsonObject = new JSONObject();
			jsonObject.put("fieldName",dataModelFieldEntities.get(i).getFieldName());
			jsonObject.put("fieldType",dataModelFieldEntities.get(i).getFieldType());
			jsonObject.put("fieldComments",dataModelFieldEntities.get(i).getFieldCnName());
			jsonObjects.add(jsonObject);
		}
		return jsonObjects;
	}

	/**
	 * 保存实时数据到数据库中 MYSQL的
	 */
	public void  saveRealTimeData(String value, long modelId, String modelName, String tableName, String structure, String qualityRule,String routeName){
		LambdaQueryWrapper<DataModelFieldEntity> wrapper = Wrappers.lambdaQuery();
		wrapper.eq(DataModelFieldEntity::getDataModelId,modelId);
		List<String> list = dataModelFieldService.list(wrapper).stream().map(DataModelFieldEntity::getFieldName).collect(Collectors.toList());
		String sql="";
		try {
			//根据模型id讲师获取保存到表
			JSONObject jsonObject = JSONObject.parseObject(value);
			if (StringUtils.isBlank(structure)) {
				//质量检验规则
				kafkaConsumerListener.qualityRuleData(modelId,modelName,qualityRule,jsonObject);
				//保存数据到数据库中
				sql=dataDisposeDubbo.generateInsertStatement(tableName,jsonObject,list);
			}else {
				//质量检验规则
				kafkaConsumerListener.qualityRuleData(modelId,modelName,qualityRule,jsonObject.getJSONObject(structure));
				//保存数据到数据库中
				sql=dataDisposeDubbo.generateInsertStatement(tableName,jsonObject,list);
			}
			//如果实时数据服务中有该数据路由，则发送数据
			LambdaQueryWrapper<DataServicesEntity> wrapper1 = Wrappers.lambdaQuery();
			wrapper1.eq(DataServicesEntity::getTargetTopic,routeName);
			List<DataServicesEntity> data = dataServices.list(wrapper1);
			//判断路由是否在服务中，如果服务中有，则发送该数据 判断是通过kafka发送还是websocket还是rabbitmq发送
			if (data!=null){
				Iterator<DataServicesEntity> iterator = data.iterator();
				if (iterator.hasNext()){
					DataServicesEntity dataServices = iterator.next();
					if ("rabbitmq".toLowerCase().equals(dataServices.getServiceType())){
						sendRabbitMqData(modelId,jsonObject,dataServices.getClusterId(),dataServices.getTopicName());
					}
				}
			}
			//执行SQL，保存数据 不创建连接
			dataDisposeDubbo.insertDataToTable(sql);
		}catch (NullPointerException e){
			log.error("消息体与模型字端不匹配，请检查消息体！！！");
		}
		catch (Exception e){
			e.printStackTrace();
		}
	}

	/**
	 * rabbitmq数据发送
	 * @param modelId
	 * @param data
	 * @param clusterId
	 * @param routingName
	 */
	private void sendRabbitMqData(long modelId, JSONObject data, Long clusterId, String routingName) {
		try {
			RabbitMQEntity mqEntity = this.getById(clusterId);
			//创建连接
			Connection connection = getConnection2(mqEntity);
			//创建通道
			Channel channel = connection.createChannel();
			List<DataModelFieldEntity> list = dataModelFieldService.getDataModelFieldList(modelId);
			Iterator<DataModelFieldEntity> iterator = list.iterator();
			JSONObject dataObject = new JSONObject();
			while (iterator.hasNext()){
				DataModelFieldEntity dataModelField = iterator.next();
				dataObject.put(dataModelField.getFieldName(),data.get(dataModelField.getFieldName()));
			}
			String message = dataObject.toString();
			// 1.交换机 2.路由 3消息的属性 4消息体
			channel.basicPublish(mqEntity.getExchangeName(),routingName,null,message.getBytes());
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * 定时任务，重连rabbitmq集群
	 */
	public void timingConnectionRouting() {
		List<RabbitMQEntity> list = list();
		for (RabbitMQEntity rabbitMQEntity : list) {
			try {
				Connection connection = getConnection2(rabbitMQEntity);
				if (ObjectUtil.isNotEmpty(connection)){
					List<TopicTaskEntity> taskEntityList = topicTaskService.list(new LambdaQueryWrapper<TopicTaskEntity>()
						.eq(TopicTaskEntity::getClusterId, rabbitMQEntity.getId()));
					for (TopicTaskEntity topicTaskEntity : taskEntityList) {
						asyncGet(rabbitMQEntity, connection, topicTaskEntity);
					}
				}else {
					rabbitMQEntity.setStatus(0);
					this.updateById(rabbitMQEntity);
					log.error("rabbimq连接失败地址{}"+rabbitMQEntity.getHost()+rabbitMQEntity.getPort());
				}
			}catch (Exception e){
				rabbitMQEntity.setStatus(0);
				this.updateById(rabbitMQEntity);
				log.error("rabbimq连接失败地址{}"+rabbitMQEntity.getHost()+rabbitMQEntity.getPort());
			}
		}
	}

	/**
	 * 创建路由信息
	 * @param jsonObject
	 * @return
	 */
	public R crateRouting(JSONObject jsonObject) {
		Long id = jsonObject.getLong("id");
		String routingName = jsonObject.getString("routingName");
		//查看当前集群下有哪些路由
		try {
			List<String> list = findRabbitmqRouting(id);
			if (!list.contains(routingName)){
				//获取到集群的实体类
				RabbitMQEntity mqEntity = this.getById(id);
				//创建连接
				Connection connection = this.getConnection2(mqEntity);
				if (connection!=null){
					Channel channel = connection.createChannel();
					String exchangeName = mqEntity.getExchangeName();
					channel.exchangeDeclare(exchangeName,"direct",true);
					//创建一个与路由同名的队列  1队列名称 2 是否持久化 3是否排他 4 是否自动删除 5其他属性
					channel.queueDeclare(routingName,true,false,false,null);
					channel.queueBind(routingName,exchangeName,routingName);
					return R.data("添加路由成功！");
				}else {
					return R.fail("连接失败！");
				}
			}else {
				return R.fail("添加失败，该路由以及存在:"+routingName);
			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * 查找当前集群mq中有哪些路由
	 * @param id
	 */
	private List<String> findRabbitmqRouting(Long id) {
		RabbitMQEntity mqEntity = this.getById(id);
		String host = mqEntity.getHost();
		String vhost = null;
		try {
			vhost = URLEncoder.encode(mqEntity.getVirtualHost(), "utf8");
		} catch (UnsupportedEncodingException e) {
			throw new RuntimeException(e);
		}
		//获取路由的地址
		String routingUrl="http://"+host+":"+mqEntity.getManagePort()+"/api/exchanges/"+vhost
			+"/"+mqEntity.getExchangeName()+"/bindings/source";
		//获取路由信息
		String rabbitMqUrl = getRabbitMqUrl(routingUrl, mqEntity);
		JSONArray jsonArray = JSON.parseArray(rabbitMqUrl);
		List<String> list = new ArrayList<>();
		for (int i = 0; i < jsonArray.size(); i++) {
			JSONObject jsonObject = jsonArray.getJSONObject(i);
			list.add(jsonObject.getString("routing_key"));
		}
		return list;
	}



	/**
	 * 获取当前rabbitmq集群有哪些路由和消息数量
	 */
	public R<Map<String, java.lang.Object>> getRabbitmqRouting(JSONObject Object) throws Exception {
		Long clusterId = Object.getLong("clusterId");
		Integer page = Object.getInteger("page");
		Integer limit = Object.getInteger("limit");
		RabbitMQEntity mqEntity = this.getById(clusterId);
		String host = mqEntity.getHost();
		String vhost = URLEncoder.encode(mqEntity.getVirtualHost(), "utf8");
		//获取路由的地址
		String routingUrl="http://"+host+":"+mqEntity.getManagePort()+"/api/exchanges/"+vhost
			+"/"+mqEntity.getExchangeName()+"/bindings/source";
		//获取路由信息
		String rabbitMqUrl = getRabbitMqUrl(routingUrl, mqEntity);
		JSONArray jsonArray = JSON.parseArray(rabbitMqUrl);
		List<JSONObject> list = new ArrayList<>();
		for (int i = 0; i < jsonArray.size(); i++) {
			JSONObject jsonObject = jsonArray.getJSONObject(i);
			//获取消息堆积数量的地址
			String messageUrl="http://"+host+":"+mqEntity.getManagePort()+"/api/queues/"+vhost+"/"+
				jsonObject.getString("routing_key");
			//获取消息的堆积信息
			String messagenNumberObj = getRabbitMqUrl(messageUrl, mqEntity);
			JSONObject job = new JSONObject();
			job.put("routingName",jsonObject.getString("routing_key"));
			JSONObject msgObject = JSON.parseObject(messagenNumberObj);
			Integer number = msgObject.getJSONObject("backing_queue_status").getInteger("len");
			job.put("messagenNumber",number);
			list.add(job);
		}
		Map<String, java.lang.Object> map = new HashMap<>();
		//分页展示路由和详细数量信息
		List<JSONObject> resultList = listPage(list, page, limit);
		map.put("resultList",resultList);
		map.put("count",list.size());
		return R.data(map);
	}

	/**
	 * 对得到的结果进行分页
	 * @param list
	 * @param page
	 * @param limit
	 * @return
	 */
	private List<JSONObject> listPage(List<JSONObject> list, Integer page, Integer limit) {
		page=(page-1)*limit;
		limit=Math.min(page+limit,list.size());
		if (page>=limit){
			return new ArrayList<>();
		}
		return list.subList(page,limit);
	}


	/**
	 * 控制台api调用
	 * @param routingUrl
	 * @param mqEntity
	 * @return
	 */
	private String getRabbitMqUrl(String routingUrl, RabbitMQEntity mqEntity) {
		// 发送http请求数据
		// 创建HttpClientBuilder
		HttpClientBuilder httpClientBuilder = HttpClientBuilder.create();
		// 设置BasicAuth
		BasicCredentialsProvider provider = new BasicCredentialsProvider();
		// Create the authentication scope
		AuthScope scope = new AuthScope(AuthScope.ANY_HOST, AuthScope.ANY_PORT, AuthScope.ANY_REALM);
		// Create credential pair，在此处填写用户名和密码
		UsernamePasswordCredentials credentials = new UsernamePasswordCredentials(mqEntity.getUsername(),mqEntity.getPassword());
		// Inject the credentials
		((BasicCredentialsProvider) provider).setCredentials(scope, credentials);
		// Set the default credentials provider
		httpClientBuilder.setDefaultCredentialsProvider(provider);
		// HttpClient
		CloseableHttpClient closeableHttpClient = httpClientBuilder.build();

		String result = "";
		HttpGet httpGet = null;
		HttpResponse httpResponse = null;
		HttpEntity entity = null;
		httpGet = new HttpGet(routingUrl);
		try {
			httpResponse = closeableHttpClient.execute(httpGet);
			entity = httpResponse.getEntity();
			if( entity != null ){
				result = EntityUtils.toString(entity);
			}
		}catch (HttpHostConnectException e){
			log.error("连接失败超时，请检查该集群地址！！！");
		} catch (ClientProtocolException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		// 关闭连接
		try {
			closeableHttpClient.close();
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
		return result;
	}

	/**
	 * 发送示例数据
	 * @param jsonObject
	 * @return
	 */
	public R sendExampleData(JSONObject jsonObject) {
		try {
			Long id = jsonObject.getLong("id");
			String routingName = jsonObject.getString("routingName");
			JSONObject dataObj = jsonObject.getJSONObject("data");
			RabbitMQEntity mqEntity = this.getById(id);
			//获取连接
			Connection connection = getConnection2(mqEntity);
			if (ObjectUtil.isNotEmpty(connection)){
				Channel channel = connection.createChannel();
				String exchangeName = mqEntity.getExchangeName();
				//创建一个与路由同名的队列  1队列名称 2 是否持久化 3是否排他 4 是否自动删除 5其他属性
				channel.queueDeclare(routingName,true,false,false,null);
				//发送示例数据
				channel.basicPublish(exchangeName,routingName,null,dataObj.toString().getBytes());
				//保存发送的数据到redis中，设置过期时间为五分钟
				RedisUtil.set(routingName,dataObj.toString(), Duration.ofDays(300));
				return R.data("发送消息示例成功！");
			}else {
				return R.fail("连接失败");
			}
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * 消费的示例数据
	 * @param jsonObject
	 * @return
	 */
	public R getExampleData(JSONObject jsonObject) {
		try {
			Long id = jsonObject.getLong("id");
			String routingName = jsonObject.getString("routingName");
			RabbitMQEntity mqEntity = this.getById(id);
			//获取连接
			Connection connection = getConnection2(mqEntity);
			if (ObjectUtil.isNotEmpty(connection)){
				Channel channel = connection.createChannel();
				String exchangeName = mqEntity.getExchangeName();
				channel.exchangeDeclare(exchangeName, "direct", true);
				//创建一个与路由同名的队列  1队列名称 2 是否持久化 3是否排他 4 是否自动删除 5其他属性
				channel.queueDeclare(routingName,true,false,false,null);
				channel.queueBind(routingName,exchangeName,routingName);
				//消费消息
				channel.basicConsume(routingName, true, new DefaultConsumer(channel) {
					@Override
					public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] body) throws IOException {
						String msg = new String(body, "UTF-8");
						channel.basicAck(envelope.getDeliveryTag(),false);
					}
				});
				//获取redis中mq发送的数据
				String result = RedisUtil.get(routingName);
				return R.data(result);
			}else {
				return R.fail("连接失败");
			}
		} catch (IOException e) {
			e.getMessage();
		}
		return null;
	}
}
